package com.moli.iov.handle.worker.his.consumer;

import com.moli.iov.consumer.KafkaBatchConsumerServer;
import com.moli.iov.handle.worker.his.util.ConsumerCommonServer;
import com.moli.iov.handle.worker.his.util.ConsumerUtil;
import com.moli.iov.protocol.codec.ReqStreamData;
import com.moli.iov.protocol.model.dto.VehicleOnLineDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.support.Acknowledgment;

import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.List;

/**
   * 登录报文处理
   * @param
   * @author lixueyong
   * @return
   */
@Slf4j
public class ConsumerOnlineHandle extends KafkaBatchConsumerServer {
    @Value("${hbase.tableName.iov.online}")
    private  String HBASE_TABLENAME_IOV_ONLINE;
    @Resource
    private ConsumerCommonServer consumerCommonServer;
    @Override
    public void manualHandle(List<ConsumerRecord<String, Object>> list, Acknowledgment acknowledgment) {
        HashMap<String, HashMap<String, Object>> onlineMap = new HashMap<>(0);
        String topic = "";
        for (ConsumerRecord record : list) {
            topic = record.topic();
            VehicleOnLineDto vehicleOnLineDto = (VehicleOnLineDto) record.value();
            if (null != vehicleOnLineDto) {
                HashMap<String, Object> map = new HashMap<String, Object>(0);
                map.put("messageContent", vehicleOnLineDto);
                long timeStamp = vehicleOnLineDto.getTimestamp();
                Date date = new Date(timeStamp);
                String rowKey = ConsumerUtil.getRowkey(record.key().toString(), date);
                onlineMap.put(rowKey, map);
            } else {
                logger.error("[车辆topic={}主题报文为空]", topic);
            }
        }
        if (!onlineMap.isEmpty()) {
            try {
                consumerCommonServer.insertBatchData(HBASE_TABLENAME_IOV_ONLINE, onlineMap);
                acknowledgment.acknowledge();
            } catch (Exception e) {
                log.error("hbase保存{}数据共{}条失败.....{}", topic, list.size(), e);
                e.printStackTrace();
            }
        }
    }

}



